package cn.com.itg.function.process;

import cn.com.itg.common.BuildVOUtil;
import cn.com.itg.pojo.domain.entity.MATDOC;
import cn.com.itg.pojo.domain.entity.ZTTCIDENTICAL;
import cn.com.itg.pojo.domain.vo.MatdocSendVO;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

public class MatdocZqrbwartBPFunction extends BroadcastProcessFunction<MATDOC, List<ZTTCIDENTICAL>, MatdocSendVO> implements Serializable {
    private List<ZTTCIDENTICAL> ztt = new ArrayList<>();
    @Override
    public void processElement(MATDOC matdoc, ReadOnlyContext readOnlyContext, Collector<MatdocSendVO> collector) throws Exception {
        ztt.stream()
            .forEach(zttcidentical -> {
            //1.BWART == ZDTYPE（*为所有）
            if (!Objects.equals(matdoc.getBwart(), zttcidentical.getZdtype())) {
                return;
            }
            //2.SHKZG = ZFIELD
            if (Objects.equals(matdoc.getShkzg(), zttcidentical.getZfield())
                || Objects.equals("*", matdoc.getShkzg())
                || Objects.equals("*", zttcidentical.getZfield())) {
                    collector.collect(BuildVOUtil.buildSendVOByZqrbwart(matdoc, zttcidentical));
            }
        });
    }

    @Override
    public void processBroadcastElement(List<ZTTCIDENTICAL> zttcidenticals, Context context, Collector<MatdocSendVO> collector) throws Exception {
        ztt = zttcidenticals;
    }


}
